package com.udf.flink.scala.apitest.broadcast
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

object ConnectBroadcast {
  def main(args: Array[String]): Unit = {
    // 获取执行流处理引擎
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 行为流 -- 非广播流
    val actionStream: KeyedStream[(String, Int), String] = env.socketTextStream("localhost",9999)
//      .addSource(new FlinkKafkaConsumer010[String]("test", new SimpleStringSchema(), initProps()))
      .map(data => {
      val arr = data.split(",")
//        ((arr(0),arr(1).toInt),1)
        (arr(0),arr(1).toInt)
      }).keyBy(_._1)

    // 广播状态描述
    val broadcastStateDesc: MapStateDescriptor[String, String] =
      new MapStateDescriptor[String, String]("broadcast-desc", classOf[String], classOf[String])

    // 规则流 -- 广播流
    val ruleStream: BroadcastStream[String] = env.socketTextStream("localhost",9000)
//      .addSource(new FlinkKafkaConsumer010[String]("test_1", new SimpleStringSchema(), initProps()))
      .broadcast(broadcastStateDesc) // 将基础流转为广播流的时候需要指定广播流的描述信息

    // 使用connect算子将 主体基本流 和 广播流连接起来
    val connectedStream: BroadcastConnectedStream[(String, Int), String] = actionStream.connect(ruleStream)

    // 处理连接流数据
    connectedStream
      .process(new MyKeyedBroadcastProcessFunction(broadcastStateDesc))
      .print()

    env.execute("broadcast_stream")
  }

}
class MyKeyedBroadcastProcessFunction(broadcastStateDesc: MapStateDescriptor[String, String])
  extends KeyedBroadcastProcessFunction[String, (String, Int), String, String] {
  // 每当 主体基本流新增一条记录，该方法就会执行一次
  override def processElement(in1: (String, Int),
                              readOnlyCtx: KeyedBroadcastProcessFunction[String, (String, Int), String, String]#ReadOnlyContext,
                              collector: Collector[String]): Unit = {
    // 从 广播状态中根据key获取数据(规则数据)
    val ruleString: String = readOnlyCtx.getBroadcastState(broadcastStateDesc).get("rule")
    collector.collect(in1 + ruleString)
  }

  // 每当 广播流新增一条记录，该方法就会执行一次
  override def processBroadcastElement(in2: String,
                                       ctx: KeyedBroadcastProcessFunction[String, (String, Int), String, String]#Context,
                                       collector: Collector[String]): Unit = {
    // 获取广播状态并更新状态数据(规则数据)
    ctx.getBroadcastState(broadcastStateDesc).put("rule", in2)
  }
}
/*
下边说明需要注意的是：我们仅对BroadcastState中key为"rule"对应的value值进行更改操作

第一步，生产数据step_01到test_1中，此时控制台没有打印信息
       但此时，我们已经将step_01这条数据存放到了broadcastState中

第二步，生产数据step_02到test中，此时控制台打印信息(step_02,1)step_01
       说明actionStream中的数据和从broadcastState捕获到规则数据拼接并打印

第三步，生产数据step_03到test中，actionStream中的数据和从broadcastState捕获到规则数据拼接并打印

第四步，生产数据step_04到test_1中，此时控制台没有打印信息
      但此时，我们已经更新了broadcastState中存放的数据为step_04

第五步，生产数据step_05到test中，根据打印信息可以得知，我们当前获取到了broadcastState中最新的值
      也就是，我们的非广播流数据实时的感知到广播流数据的流动情况
 */